package com.company.project.service.impl;

import com.alibaba.fastjson.JSON;
import com.company.project.constant.MQEnums;
import com.company.project.constant.TradeEnums;
import com.company.project.dao.TradeMqProducerTempMapper;
import com.company.project.dao.TradePayMapper;
import com.company.project.exception.AceMQException;
import com.company.project.model.TradeMqProducerTemp;
import com.company.project.model.TradePay;
import com.company.project.protocol.mq.PaidMQ;
import com.company.project.protocol.pay.CallbackPaymentReq;
import com.company.project.protocol.pay.CallbackPaymentRes;
import com.company.project.protocol.pay.CreatePaymentReq;
import com.company.project.protocol.pay.CreatePaymentRes;
import com.company.project.rocketmq.AceMQProducer;
import com.company.project.service.TradePayService;
import com.company.project.core.AbstractService;
import com.company.project.util.IDGenerator;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import tk.mybatis.mapper.entity.Condition;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * Created by CodeGenerator on 2018/12/07.
 */
@Service
@Transactional
public class TradePayServiceImpl extends AbstractService<TradePay> implements TradePayService {

    @Resource
    private TradePayMapper tradePayMapper;

    /*
     * 消息生产者 第三方支付平台回调我们之后，修改订单状态为已支付。解耦合
     */
    @Resource
    private AceMQProducer aceMQProducer;
    /*
     * 用于 发送可靠消息
     */
    @Resource
    private TradeMqProducerTempMapper tradeMqProducerTempMapper;

    private ExecutorService executorService = Executors.newFixedThreadPool(10);

    public CreatePaymentRes createPayment(CreatePaymentReq createPaymentReq) {
        CreatePaymentRes createPaymentRes = new CreatePaymentRes();
        createPaymentRes.setRetCode(TradeEnums.RetEnum.SUCCESS.getCode());
        createPaymentRes.setRetInfo(TradeEnums.RetEnum.SUCCESS.getDesc());
        try {
            Condition c = new Condition(TradePay.class);
            c.createCriteria()
                    .andEqualTo("orderId",createPaymentReq.getOrderId())
                    .andEqualTo("isPaid",TradeEnums.YesNoEnum.YES.getCode());
            long count = this.tradePayMapper.selectCountByCondition(c);
            if (count > 0) {
                throw new Exception("该订单已支付");
            }
            String payId = IDGenerator.generatorUUID();
            TradePay tradePay = new TradePay();
            tradePay.setPayId(payId);
            tradePay.setOrderId(createPaymentReq.getOrderId());
            tradePay.setIsPaid(TradeEnums.YesNoEnum.NO.getCode());
            tradePay.setPayAmount(createPaymentReq.getPayAmount());
            tradePayMapper.insert(tradePay);
            System.out.println("创建支付订单成功:" + payId);
        } catch (Exception e) {
            createPaymentRes.setRetCode(TradeEnums.RetEnum.FAIL.getCode());
            createPaymentRes.setRetInfo(e.getMessage());
        }
        return createPaymentRes;
    }

    /*
     * 回调支付接口，修改状态。发送可靠消息
     */
    @Transactional
    public CallbackPaymentRes callbackPayment(CallbackPaymentReq callbackPaymentReq) {
        CallbackPaymentRes callbackPaymentRes = new CallbackPaymentRes();
        callbackPaymentRes.setRetCode(TradeEnums.RetEnum.SUCCESS.getCode());
        callbackPaymentRes.setRetInfo(TradeEnums.RetEnum.SUCCESS.getDesc());

        // 这里屏蔽了 参数校验 验证签名什么的
        //TODO
        if (callbackPaymentReq.getIsPaid().equals(TradeEnums.YesNoEnum.YES.getCode())) {
            //更新支付状态
            TradePay tradePay = tradePayMapper.selectByPrimaryKey(callbackPaymentReq.getPayId());
            if (tradePay == null) {
                throw new RuntimeException("未找到支付订单");
            }
            if (tradePay.getIsPaid().equals(TradeEnums.YesNoEnum.YES.getCode())) {
                throw new RuntimeException("该支付单已支付");
            }
            tradePay.setIsPaid(TradeEnums.YesNoEnum.YES.getCode());
            int i = tradePayMapper.updateByPrimaryKeySelective(tradePay);
            /*
             * 当前rocketMQ不支持分布式事务，不能保证修改数据库和MQ消息同时成功或失败。 把MQ事务变成本地事务 --
             * trade_mq_producer_temp
             */
            if (i == 1) {
                final PaidMQ paidMQ = new PaidMQ();
                paidMQ.setPayAmount(tradePay.getPayAmount());
                paidMQ.setOrderId(tradePay.getOrderId());
                paidMQ.setPayId(tradePay.getPayId());
                //消息发送临时表
                final TradeMqProducerTemp mqProducerTemp = new TradeMqProducerTemp();
                mqProducerTemp.setId(IDGenerator.generatorUUID());
                mqProducerTemp.setGroupName("payProducerGroup");
                mqProducerTemp.setMsgKeys(tradePay.getPayId());
                mqProducerTemp.setMsgTopic(MQEnums.TopicEnum.PAY_PAID.getTopic());
                mqProducerTemp.setMsgTag(MQEnums.TopicEnum.PAY_PAID.getTag());
                mqProducerTemp.setMsgBody(JSON.toJSONString(paidMQ));
                mqProducerTemp.setCreateTime(new Date());
                tradeMqProducerTempMapper.insert(mqProducerTemp);
                // 异步线程发送mq，发送成功清空发送表
                // 如果未发送成功，，用定时任务 根据创建时间 去扫描这张表 用补偿的方式  发送消息
                executorService.submit(new Runnable() {
                    public void run() {
                        try {
                            SendResult sendResult = aceMQProducer.sendMessage(MQEnums.TopicEnum.PAY_PAID,
                                    paidMQ.getPayId(), JSON.toJSONString(paidMQ));
                            System.out.println(sendResult);
                            if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                                tradeMqProducerTempMapper.deleteByPrimaryKey(mqProducerTemp.getId());
                                System.out.println("删除消息表成功");
                            }
                        } catch (AceMQException e) {
                            e.printStackTrace();
                        }
                    }
                });
            } else {
                throw new RuntimeException("该支付单已支付");
            }
        }
        return callbackPaymentRes;
    }

}


